#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include "ThreadPool.h"

const int NUMBER = 2;				//一次性创建或销毁多少个线程

//	任务结构体
typedef struct Task {
	void(*funtion)(void* arg);		//	利用泛型编程实现
	void* arg;						//	数据
}Task; 

//	线程池结构体
struct ThreadPool {
	
	Task* task_queue;				//	任务队列
	int queue_capacity;				//	容量
	int queue_size;					//	当前任务个数
	int queue_front;				//	队头
	int queue_rear;					//	队尾

	pthread_t manager_id;			//	管理者线程id
	pthread_t* thread_ids;			//	工作的线程id
	int min_num;					//	最少线程数量
	int max_num;					//	最多线程数量
	int busy_num;					//	正在工作的线程个数
	int live_num;					//	存活的线程个数
	int exit_num;					//	要销毁的线程个数

	pthread_mutex_t mx_pool;		//	锁住线程池
	pthread_mutex_t mx_busy;		//	锁 busy_num 变量
	pthread_cond_t not_full;		//	条件变量：用来判断任务队列是否为满
	pthread_cond_t not_empty;		//	条件变量：用来判断任务队列是否为空

	int shut_down;					//	是否需要销毁线程池	1：销毁	0：不销毁
};

//	线程池初始化函数
ThreadPool* thread_pool_create(int min, int max, int size)
{

	ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
	do {
		if (pool == NULL) {				//	判断是否成功申请到内存空间
			printf("线程池分配空间失败！");
			break;
		}

		pool->thread_ids = (pthread_t*)malloc(sizeof(pthread_t) * max);
		if (pool->thread_ids == NULL) {
			printf("线程id内存空间分配失败！");
			break;
		}

		//	初始化
		memset(pool->thread_ids, 0, sizeof(pthread_t) * max);
		pool->min_num = min;
		pool->max_num = max;
		pool->busy_num = 0;
		pool->live_num = min;		//活着的线程应该等于最小的线程数量
		pool->exit_num = 0;

		//	初始化锁和条件变量
		if (pthread_mutex_init(&pool->mx_pool, NULL) != 0
			|| pthread_mutex_init(&pool->mx_busy, NULL) != 0
			|| pthread_cond_init(&pool->not_full, NULL) != 0
			|| pthread_cond_init(&pool->not_empty, NULL) != 0) 
		{
			printf("初始化锁活着条件变量时失败！");
			break;
		}

		//	初始化任务队列
		pool->task_queue = (Task*)malloc(sizeof(Task) * size);
		pool->queue_capacity = size;
		pool->queue_size = 0;
		pool->queue_front = 0;
		pool->queue_rear = 0;
		pool->shut_down = 0;

		//	创建线程
		pthread_create(&pool->manager_id, NULL, manager, pool);
		for (int i = 0; i < min; i++) 
		{
			pthread_create(&pool->thread_ids[i], NULL, worker, pool);
		}
		return pool;
	} while (0);		//使用do while(0)的原因是因为后面需要用以释放内存资源

	//	释放资源
	if (pool && pool->thread_ids)	free(pool->thread_ids);
	if (pool && pool->task_queue)	free(pool->task_queue);
	if (pool)	free(pool);

	return NULL;
}

//	工作函数
void* worker(void* arg) {
	ThreadPool* pool = (ThreadPool*)arg;

	while (1) 
	{
		pthread_mutex_lock(&pool->mx_pool);				//	上锁

		//	当前任务队列是否为空
		while (pool->queue_size == 0 && !pool->shut_down) 
		{
			//	为空就阻塞工作线程
			pthread_cond_wait(&pool->not_empty, &pool->mx_pool);

			//判断是不是要销毁线程
			if (pool->exit_num > 0) 
			{
				pool->exit_num--;
				if (pool->live_num > pool->min_num) 
				{
					pool->live_num--;
					pthread_mutex_unlock(&pool->mx_pool);	//	记得解锁
					thread_exit(pool);
				}
			}
		}

		//	判断线程池是否被关闭了
		if (pool->shut_down)
		{
			pthread_mutex_unlock(&pool->mx_pool);		//	解锁
			thread_exit(pool);							//	退出线程
		}

		//	从任务队列中取出一个任务
		Task task;
		task.funtion = pool->task_queue[pool->queue_front].funtion;
		task.arg = pool->task_queue[pool->queue_front].arg;
		//	出队
		pool->queue_front = (pool->queue_front + 1) % pool->queue_capacity;
		pool->queue_size--;

		//	任务队列有位置了，唤醒生产者
		pthread_cond_signal(&pool->not_full);
		pthread_mutex_unlock(&pool->mx_pool);			//	解锁

		printf("线程 %ld 开始工作...", pthread_self());
		pthread_mutex_lock(&pool->mx_busy);
		pool->busy_num++;
		pthread_mutex_unlock(&pool->mx_busy);

		//	执行任务
		task.funtion(task.arg);
		free(task.arg);
		task.arg = NULL;

		printf("线程 %ld 结束工作", pthread_self());
		pthread_mutex_lock(&pool->mx_busy);
		pool->busy_num--;
		pthread_mutex_unlock(&pool->mx_busy);
	}
	
	return NULL;
}

//	管理者线程工作函数
void* manager(void* arg) {
	ThreadPool* pool = (ThreadPool*)arg;
	while (!pool->shut_down) 
	{
		//	每隔三秒检测一次
		sleep(3);

		//	取出线程池中任务的数量和当前工作的线程数量
		pthread_mutex_lock(&pool->mx_pool);
		int queue_size = pool->queue_size;
		int live_num = pool->live_num;
		pthread_mutex_unlock(&pool->mx_pool);

		//	取出正在工作的线程的数量
		pthread_mutex_lock(&pool->mx_busy);
		int busy_num = pool->busy_num;
		pthread_mutex_unlock(&pool->mx_busy);

		//	添加线程	根据实际的业务要求来制定
		//	规则：任务的个数>存活的线程的个数 && 存活的线程数<最大线程数
		if (queue_size > live_num && live_num < pool->max_num) 
		{
			pthread_mutex_lock(&pool->mx_pool);			//	使用线程池元素时记得加锁
			int count = 0;
			//	遍历线程id组找出没有使用的线程id来创建线程
			for (int i = 0; i < pool->max_num && count < NUMBER
				&& pool->live_num < pool->max_num; i++) 
			{
				if (pool->thread_ids[i] == 0) 
				{
					pthread_create(&pool->thread_ids[i], NULL, worker, pool);
					count++;
					pool->live_num++;
				}
			}
			pthread_mutex_unlock(&pool->mx_pool);
		}

		//	销毁线程	根据实际的业务要求来制定规则
		//	规则：正在工作的线程 * 2 < 存活的线程数	&& 存活的线程 > 最小线程数
		if (busy_num * 2 < live_num&&live_num > pool->min_num) 
		{
			pthread_mutex_lock(&pool->mx_pool);
			pool->exit_num = NUMBER;
			pthread_mutex_unlock(&pool->mx_pool);

			//	让工作线程自杀
			for (int i = 0; i < NUMBER; i++) 
			{
				pthread_cond_signal(&pool->not_empty);		//	唤醒线程

			}
		}
	}
	return NULL;
}

//	线程退出函数
void thread_exit(ThreadPool* pool) {
	//获取当前线程id
	pthread_t tid = pthread_self();
	for (int i = 0; i < pool->max_num; i++) 
	{
		if (pool->thread_ids[i] == tid)
		{
			pool->thread_ids[i] = 0;
			printf("线程 %ld 已销毁", tid);
			break;
		}
	}
	pthread_exit(NULL);
}

//	给线程池添加任务
void thread_pool_add_task(ThreadPool* pool, void(*func)(void*), void* arg) 
{
	pthread_mutex_lock(&pool->mx_pool);				//	给线程池上锁
	//	判断任务队列是否已满
	while (pool->queue_size == pool->queue_capacity && !pool->shut_down) 
	{
		//	阻塞生产者线程
		pthread_cond_wait(&pool->not_full, &pool->mx_pool);
	}
	if (pool->shut_down) 
	{
		pthread_mutex_unlock(&pool->mx_pool);
		return;
	}
	//	添加任务
	pool->task_queue[pool->queue_rear].funtion = func;
	pool->task_queue[pool->queue_rear].arg = arg;
	pool->queue_rear = (pool->queue_rear + 1) % pool->queue_capacity;
	pool->queue_size++;

	//	唤醒阻塞线程来完成任务
	pthread_cond_signal(&pool->not_empty);
	pthread_mutex_unlock(&pool->mx_pool);			//	给线程池解锁
}

//	获取线程池中工作的线程的个数
int get_thread_pool_busy_num(ThreadPool* pool) {
	pthread_mutex_lock(&pool->mx_busy);
	int res = pool->busy_num;
	pthread_mutex_unlock(&pool->mx_busy);
	return res;
}

//	获取线程池中存活的线程个数
int get_thread_pool_live_num(ThreadPool* pool) {
	pthread_mutex_lock(&pool->mx_pool);
	int res = pool->live_num;
	pthread_mutex_unlock(&pool->mx_pool);
	return res;
}

//	销毁线程池
int thread_pool_destroy(ThreadPool* pool) {
	if (pool == NULL) 
	{
		return -1;
	}

	//	关闭线程池
	pool->shut_down = 1;

	//	阻塞回收管理者线程
	pthread_join(pool->manager_id, NULL);

	//	唤醒阻塞消费者线程
	for (int i = 0; i < pool->live_num; i++) {
		pthread_cond_signal(&pool->not_empty);
	}

	//	释放堆内存
	if (pool->task_queue)	free(pool->task_queue);
	if(pool->thread_ids)	free(pool->thread_ids);
	pthread_mutex_destroy(&pool->mx_pool);
	pthread_mutex_destroy(&pool->mx_busy);
	pthread_cond_destroy(&pool->not_empty);
	pthread_cond_destroy(&pool->not_full);
	//由于前面判断过这个pool是不为空的，所以这里不用判断
	free(pool);
	pool = NULL;
	return 0;
}